import 'dart:async';
import 'dart:io';

import 'package:flutter/foundation.dart';

void main() {
  print("main start");
  // dart只有一个线程，这里sleep直接将dart的唯一一个主线程睡眠了，后面所有任务都会阻塞，最好不要用这个方法。
  sleep(const Duration(seconds: 1));
  // periodicTest();

  // fromFutureTest();

  // fromFuturesTest();
  // listenStreamTest();
  streamControllerTest();
  testStreamSubscription();
  print("main end");
}

// 该种方式与EventBus功能一致，但不是回调，而是stream
void testStreamSubscription() {
  // 注：这里必须是broadcast才能多订阅者，普通stream是单订阅模式
  // 在service中可以加get方法
  StreamController<Model> _streamController = StreamController.broadcast();
  // 在service中可以加get方法
  Stream<Model> _stream = _streamController.stream;

  // 订阅者之一，通常在initState中初始化，在dispose中cancel
  StreamSubscription streamSubscription = _stream.listen((model) {
    debugPrint(model.toString());
  });
  Future.delayed(const Duration(seconds: 10)).then((value) {
    // 在dispose方法中取消
    streamSubscription.cancel();
  });
  // 通过get获取stream源
  // .sink.add is no difference from .add call.
  _streamController.sink.add(Model(iId: 11, name: "aa", age: 1));
  // streamSubscription.cancel(); in dispose
}

streamControllerTest() {
  // 创建
  StreamController streamController = StreamController(
      onListen: () => print("onListen"),
      onPause: () => print("onPause"),
      onResume: () => print("onResume"),
      onCancel: () => print("onCancel"),
      sync: false);
  // 放入事件
  streamController.add('element_1');
  streamController.addError("this is error");
  streamController.sink.add('element_2');
  // 监听事件
  streamController.stream.listen(
    print,
    onError: print,
    onDone: () => print("onDone"),
  );
}

listenStreamTest() async {
  Stream<String> stream = Stream.periodic(const Duration(seconds: 1), (int value) {
    return "$value秒";
  });

  await for (var s in stream) {
    print(s);
    if (s == "2秒") {
      break;
    }
  }

  stream.listen((event) {
    print(event);
  });

  stream.forEach((element) {
    print(element);
  });
}

fromFuturesTest() async {
  Future<String> fut1 = Future(() async {
    await Future.delayed(const Duration(seconds: 3));
    return "async task1 need 3 seconds";
  });
  Future<String> fut2 = Future(() async {
    await Future.delayed(const Duration(seconds: 4));
    return "async task2 need 4 seconds.";
  });

  // 将多个Future放入一个列表中，将该列表传入
  Stream<String> stream = Stream<String>.fromFutures([fut1, fut2]);
  await for (var s in stream) {
    print(s);
  }
}

fromFutureTest() async {
  Future<String> fut = Future(() async {
    await Future.delayed(const Duration(seconds: 1));
    return "fromFutureResult need 1 second";
  });

  // 该方法从一个Future创建Stream，当Future执行完成时，就会放入Stream中，而后从Stream中将任务完成的结果取出
  // 从Future创建Stream
  Stream<String> stream = Stream<String>.fromFuture(fut);
  await for (var s in stream) {
    print(s);
  }
}

periodicTest() async {
  // 要注意，这个流是无限的，它没有任何一个约束条件使之停止。在后面会介绍如何给流设置条件。
  Stream<String> stream = Stream.periodic(const Duration(seconds: 1), (int value) {
    return "$value秒";
  });

  await for (var s in stream) {
    print(s);
    if (s == "2秒") {
      break;
    }
  }
}

class Model {
  int iId;
  String name;
  int age;

  Model({required this.iId, required this.name, required this.age});

  @override
  String toString() {
    return 'Model{iId: $iId, name: $name, age: $age}';
  }
}
